package com.atguigu.flink.chapter02_DataStreamAPI.simpletrans;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;

/**
 * Created by Smexy on 2022/10/22
 *
 *  处理元素时，需要连接数据库。
 *          处理一个元素，获取一个连接?
 *          还是一个并行度，一个并行度，获取一个连接?
 *                  选择这个，连接是可以复用的。 创建连接是一个很耗时，耗资源的操作。
 *
 *          Spark：
 *                  RDD.map(x -> fun(x))   :   一个元素调用一次函数处理一次
 *                  RDD.mapPartitions(partition -> fun(partition)): 以分区为单位进行处理
 *
 *
 *    -------------------------------
 *    class RichMapFunction<IN, OUT> extends AbstractRichFunction
 *         implements MapFunction<IN, OUT>
 *
 *             RichMapFunction是MapFunction的子类，拥有 AbstractRichFunction的方法！
 *                  在AbstractRichFunction，有open(),close()
 *
 *         几乎所有的算子，都提供了 RichXxx版本！
 *
 *
 */
public class Demo6_MapRich
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        DataStreamSource<Integer> source = env.fromElements(1, 2, 3);

        source.map(new RichMapFunction<Integer, String>()
        {
            Connection conn = null;
            //并行度创建时，获取一个数据库的连接
            @Override
            public void open(Configuration parameters) throws Exception {
                //conn = Driver...
                System.out.println("连接创建好了");
            }

            //并行度关闭时，释放连接
            @Override
            public void close() throws Exception {
                System.out.println("连接释放了");
            }

            @Override
            public String map(Integer value) throws Exception {
                return value + "是数字";
            }
        }).print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
